package com.mall.order.rocketmq;

import cn.hutool.db.Db;
import cn.hutool.db.Entity;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.sql.SQLException;
import java.util.List;

/**
 * Created by IntelliJ IDEA.
 *
 * @Author: likelu
 * @Date: 2021/4/29 17:06
 * @Description:
 */
@RequestMapping("/rocketMq1")
@RestController
public class RobackConsumer {

    public String rocketAddr = "106.14.33.182:9876";

    public void consumer() {
        try {
            // 设置消费者组
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group-roback");
            consumer.setVipChannelEnabled(false);
            consumer.setNamesrvAddr(rocketAddr);
            // 设置消费者端消息拉取策略，表示从哪里开始消费
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

            // 设置消费者拉取消息的策略，*表示消费该topic下的所有消息，也可以指定tag进行消息过滤
            consumer.subscribe("TopicOrderRoback", "Roback");

            // 消费者端启动消息监听，一旦生产者发送消息被监听到，就打印消息，和rabbitmq中的handlerDelivery类似
            consumer.registerMessageListener(new MessageListenerConcurrently() {

                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    for (MessageExt messageExt : msgs) {
                        String topic = messageExt.getTopic();
                        String tag = messageExt.getTags();
                        String userId = new String(messageExt.getBody());
                        try {
                            List<Entity> entity = Db.use().query("select * from `order_history` where user='"+userId+"'");
                            for(Entity vo:entity){
                                switch (vo.get("type").toString()){
                                    case "1":
                                        Db.use().execute("update user set point=point+"+Integer.valueOf(vo.get("content").toString())
                                                +" where id='"+ vo.get("user") + "'");
                                        break;
                                    case "2":
                                        Db.use().execute("update user set balance=balance+"+Integer.valueOf(vo.get("content").toString())
                                                +" where id='"+ vo.get("user") + "'");
                                        break;
                                    case "3":
                                        Db.use().execute("update coupon set status=0 where user_id='"+ vo.get("user") + "'");
                                        break;
                                }
                                Db.use().execute("update `order` set status=0 where user_id='"+ vo.get("user") + "'");
                            }
                        }catch (Exception e){
                            e.printStackTrace();
                        }
                        System.out.println("异常回滚-执行结束");
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });

            // 调用start()方法启动consumer
            consumer.start();
            consumer.createTopic("broker-a","TopicOrderRoback",10);
            System.out.println("Consumer2 Started....");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


}
